package com.sunday.common.mq.rabbit.study.base.e9_TimeToLive_Expiration;

import com.rabbitmq.client.*;
import com.sunday.common.mq.rabbit.study.base.ConnectionUtil;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.junit.jupiter.api.Test;

import java.nio.charset.StandardCharsets;
import java.util.Collections;
import java.util.HashMap;
import java.util.concurrent.TimeUnit;

/**
 * https://www.rabbitmq.com/ttl.html
 * 生存时间和到期时间 （所有例子需要借助死信交换能力）
 * <p>
 * RabbitMQ 允许您为消息和队列设置 TTL（生存时间）。过期 消息和队列将被删除：细节将更详细地介绍 本指南的后面部分。
 * TTL 行为由可选的队列参数控制，最好使用策略来完成。
 * 消息 TTL 可以应用于单个队列、一组 队列或逐条消息应用。
 * TTL 设置也可以由操作员策略强制执行。
 */
@Slf4j
public class TimeToLiveAndExpiration {

    /**
     * 队列中的每队列消息 TTL
     * 通过使用策略设置 message-ttl 参数或在队列声明时指定相同的参数，可以为给定队列设置消息 TTL。
     * 在队列中存在的时间超过配置的 TTL 的消息被称为 死信。请注意，路由到多个队列的消息 可以在不同的时间死亡，或者根本不死亡，在每个队列中 它所在的位置。一个队列中的消息死亡没有 对其他队列中同一消息的生存期的影响。
     * 服务器保证不会传递死消息 使用 basic.deliver（给消费者）或包含在 basic.get-ok 响应中 （用于一次性提取操作）。 此外，服务器将尝试在或 在基于 TTL 的到期后不久。
     * TTL 参数或策略的值必须是 非负整数 （0 <= n）， 以毫秒为单位描述 TTL 周期。因此，值为 1000 表示添加到队列的消息将驻留在 排队 1 秒或直到它交付给使用者。
     */
    @Test
    @SneakyThrows
    public void perQueueMessageTTLInQueues() {

        String EXCHANGE_NAME = "fanout_exchange_multi_queue";
        String DEAD_EXCHANGE_NAME = "dead_direct_exchange";
        String DEAD_ROUTING_KEY_NAME = "dead_direct_routing_key";
        String DEAD_QUEUE_NAME = "dead_direct_queue";

        String queue = "multi_queue_";

        int ttl = 3000;

        ConnectionFactory cf = ConnectionUtil.factory();
        try (Connection connection = cf.newConnection()) {
            Channel channel = connection.createChannel();
            // ttl队列
            channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.FANOUT);
            // 死信交换
            channel.exchangeDeclare(DEAD_EXCHANGE_NAME, BuiltinExchangeType.DIRECT);
            channel.queueDeclare(DEAD_QUEUE_NAME, false, false, false, null);
            channel.queueBind(DEAD_QUEUE_NAME, DEAD_EXCHANGE_NAME, DEAD_ROUTING_KEY_NAME);

            for (int i = 0; i < ttl / 1000; i++) {
                channel.queueDelete(queue + i);
                final long message_ttl = ttl - (i * 1000);
                channel.queueDeclare(queue + i, false, false, false,
                        new HashMap<String, Object>() {{
                            // TTL 参数或策略的值必须是 非负整数 （0 <= n）， 以毫秒为单位描述 TTL 周期。
                            // 因此，值为 1000 表示添加到队列的消息将驻留在 排队 1 秒或直到它交付给使用者。
                            put("x-message-ttl", message_ttl);
                            put("x-dead-letter-exchange", DEAD_EXCHANGE_NAME);
                            put("x-dead-letter-routing-key", DEAD_ROUTING_KEY_NAME);
                        }});
                channel.queueBind(queue + i, EXCHANGE_NAME, "");
            }

            channel.basicPublish(EXCHANGE_NAME, "", null, EXCHANGE_NAME.getBytes(StandardCharsets.UTF_8));

            channel.basicConsume(DEAD_QUEUE_NAME, true,
                    (consumerTag, delivery) -> {
                        String message = new String(delivery.getBody(), StandardCharsets.UTF_8);
                        log.info(" consumerTag : {} ", consumerTag);
                        log.info(" envelope : {} ", delivery.getEnvelope());
                        log.info(" properties : {} ", delivery.getProperties());
                        log.info(" [x] Received '{}'", message);
                    },
                    consumerTag -> {
                    });

            /**
             * 本轮测试可知 :
             * 在 广播模式 下，绑定多个队列，分别给与队列不同的过期时间，信息的过期和死信队列的获取都不受到影响
             */
            TimeUnit.MILLISECONDS.sleep(ttl + 1000);

        }

    }


    /**
     * 发布服务器中的每消息 TTL
     * 通过在发布消息时设置过期属性，可以基于每条消息指定 TTL。
     * 过期字段的值描述 TTL 周期（以毫秒为单位）。与 x-message-ttl 相同的约束适用。
     * 由于过期字段必须是字符串，因此代理 将（仅）接受数字的字符串表示形式。
     * 当同时指定每队列和每消息 TTL 时， 将选择两者之间的较低值。
     * <p>
     * new AMQP.BasicProperties.Builder()
     * .expiration("1000")
     * .build()
     *
     * <p>
     * 警告
     * 应用了每条消息 TTL 的队列 追溯（当他们已经有消息时）将丢弃 发生特定事件时的消息。仅当过期时 消息到达队列的头部，它们实际上是 丢弃（或死信）。消费者不会有 传递给他们的过期邮件。请记住 消息过期之间可能存在自然争用条件 和消费者交付，例如消息可能过期 在它被写入套接字之后但在它到达之前 消费者。
     * 设置每条消息的 TTL 过期消息可以排队 在未过期的后面，直到后者被消耗或 过期。因此，此类过期消息使用的资源将 不会被释放，它们将被计入队列统计信息中 （例如，队列中的消息数）。
     * 以追溯方式应用每条消息的 TTL 策略时，它是 建议让消费者上网，以确保 消息被丢弃的速度更快。
     * 鉴于现有消息 TTL 设置的这种行为 队列，当需要删除消息以释放时 资源出现，应改用队列 TTL（或队列 清除或队列删除）。
     */
    @Test
    @SneakyThrows
    public void perMessageTTLInPublishers() {

        String EXCHANGE_NAME = "fanout_exchange_multi_queue";
        String DEAD_EXCHANGE_NAME = "dead_direct_exchange";
        String DEAD_ROUTING_KEY_NAME = "dead_direct_routing_key";
        String DEAD_QUEUE_NAME = "dead_direct_queue";

        String queue = "multi_queue_";

        int ttl = 3000;

        ConnectionFactory cf = ConnectionUtil.factory();
        try (Connection connection = cf.newConnection()) {
            Channel channel = connection.createChannel();

            channel.queueDelete(queue);

            // ttl队列
            channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.FANOUT);
            // 死信交换
            channel.exchangeDeclare(DEAD_EXCHANGE_NAME, BuiltinExchangeType.DIRECT);
            channel.queueDeclare(DEAD_QUEUE_NAME, false, false, false, null);
            channel.queueBind(DEAD_QUEUE_NAME, DEAD_EXCHANGE_NAME, DEAD_ROUTING_KEY_NAME);

            channel.queueDeclare(queue, false, false, false,
                    new HashMap<String, Object>() {{
                        put("x-message-ttl", ttl);
                        put("x-dead-letter-exchange", DEAD_EXCHANGE_NAME);
                        put("x-dead-letter-routing-key", DEAD_ROUTING_KEY_NAME);
                    }});
            channel.queueBind(queue, EXCHANGE_NAME, "");

            channel.basicPublish(EXCHANGE_NAME, "",
                    new AMQP.BasicProperties.Builder()
                            .expiration(String.valueOf(1000))
                            .build()
                    , (1000 + "----").getBytes(StandardCharsets.UTF_8));

            log.info("执行命令 message : {} , ttl : {}", "1000", "1000");

            for (int i = 0; i < ttl / 1000; i++) {

                channel.basicPublish(EXCHANGE_NAME, "",
                        new AMQP.BasicProperties.Builder()
                                .expiration(String.valueOf(ttl - (i * 1000)))
                                .build()
                        , ((ttl - (i * 1000)) + "----").getBytes(StandardCharsets.UTF_8));

                log.info("执行命令 message : {} , ttl : {}", ((ttl - (i * 1000)) + "----"), ttl - (i * 1000));

                channel.basicPublish(EXCHANGE_NAME, "",
                        new AMQP.BasicProperties.Builder()
                                .expiration(String.valueOf(ttl + (i * 1000)))
                                .build()
                        , ((ttl + (i * 1000)) + "++++").getBytes(StandardCharsets.UTF_8));

                log.info("执行命令 message : {} , ttl : {}", ((ttl + (i * 1000)) + "----"), ttl + (i * 1000));

            }

            channel.basicPublish(EXCHANGE_NAME, "",
                    new AMQP.BasicProperties.Builder()
                            .expiration("10000")
                            .build()
                    , "10s".getBytes(StandardCharsets.UTF_8));

            log.info("执行命令 message : {} , ttl : {}", "10s", "10000");


            channel.basicPublish(EXCHANGE_NAME, "",
                    new AMQP.BasicProperties.Builder()
                            .expiration("20000")
                            .build()
                    , "20s".getBytes(StandardCharsets.UTF_8));

            log.info("执行命令 message : {} , ttl : {}", "20s", "20000");

            channel.basicConsume(DEAD_QUEUE_NAME, true,
                    (consumerTag, delivery) -> {
                        String message = new String(delivery.getBody(), StandardCharsets.UTF_8);
                        log.info(" consumerTag : {} ", consumerTag);
                        log.info(" envelope : {} ", delivery.getEnvelope());
                        log.info(" properties : {} ", delivery.getProperties());
                        log.info(" [x] Received '{}'", message);
                    },
                    consumerTag -> {
                    });

            /**
             * 本轮测试可知 :
             * 1. 如果同时制定了Message TTL，Queue TTL，则小的那个时间生效。
             * 简单来说就是 Queue TTL 设置以后， Message TTL > Queue TTL = Queue TTL 为准，Message TTL < Queue TTL = Message TTL 为准
             * 2. 当每个Message TTL不同的时候，队列依然是按照 FIFO的规则进行，不会说某一个ttl小，就会提前被释放
             * 例如 A TTL 10 , B TTL 2 , C TTL 10 => B 的执行，是需要等到 A 过期以后才能被触发
             * 3. 补充一点，现有社区版本的确可以直接做到延迟队列的效果，不需要设置死信的方式，但是也同样收到第2点的约束，并且效率上还不如原生的
             * 当前这种情况，一但发生阶梯的过期时间的时候，就会导致消息已经过期，但是没有办法立即消费的情况
             */
            TimeUnit.MILLISECONDS.sleep(ttl * 2 + 1000);

        }

    }

    /**
     * 队列 TTL
     * TTL 也可以在队列上设置，而不仅仅是队列内容。 此功能可与自动删除队列属性一起使用。
     * 在队列上设置 TTL（过期）通常只有意义 对于暂时性（非持久）经典队列。流 不支持过期。
     * 队列仅在一段时间后过期，只有当它们 未使用（如果队列具有联机使用者，则使用队列）。
     * 通过将 x-expires 参数设置为 queue.declare， 或通过设置过期策略。此控件 队列在自动使用之前可以闲置多长时间 删除。未使用表示队列没有使用者，即 队列最近未重新声明（重新声明续订租约）， 并且 basic.get 至少在到期期间未被调用 时期。例如，这可以用于 RPC 样式的回复 队列，可以创建许多可能永远不会创建的队列 排水。
     * 服务器保证队列将被删除，如果 至少在到期期间未使用。不保证是 给出在 过期期限已过。
     * x-expires 参数或过期策略的值描述了 毫秒。它必须是正整数（与消息不同 TTL 它不能为 0）。因此，值 1000 表示队列 未使用的 1 秒将被删除。
     */
    @Test
    @SneakyThrows
    public void queueTTL() {

        ConnectionFactory cf = ConnectionUtil.factory();
        try (Connection connection = cf.newConnection()) {
            Channel channel = connection.createChannel();
            /**
             * x-expires
             */
            channel.queueDeclare("queue_ttl", false, false, false,  Collections.singletonMap("x-expires", 10000));
        }

        /**
         * 本轮测试可知 :
         * 队列将在指定的 x-expires 时间后丢失
         */
    }


    @Test
    @SneakyThrows
    public void perMessageTTLAndQueueNotTTL() {

        String QUEUE_NAME = "general_queue";

        ConnectionFactory cf = ConnectionUtil.factory();
        try (Connection connection = cf.newConnection()) {
            Channel channel = connection.createChannel();

            channel.queueDelete(QUEUE_NAME);

            channel.queueDeclare(QUEUE_NAME, false, false, false, null);

            channel.basicPublish("", QUEUE_NAME,
                    new AMQP.BasicProperties.Builder()
                            .expiration(String.valueOf(10000))
                            .build()
                    , (1000 + "----").getBytes(StandardCharsets.UTF_8));

            log.info("执行命令 message : {} , ttl : {}", "1000", "1000");

            /**
             * 本轮测试可知 :
             * Message TTL，Queue TTL 都可以独立设置，并且都会起到作用
             * 当满足过期时间后，将丢失
             */

        }

    }

}
